package psn.zjk.spring.rabbitmq.client;

import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import psn.zjk.spring.rabbitmq.MqHandler;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * @author zhangjk
 * @CREATE_TIME 2020/11/25 17:09
 */
public class MqClient {
    public static final Logger LOGGER = LoggerFactory.getLogger(MqClient.class);

    private Channel channel;

    public MqClient() {
        try {
            init();
        } catch (Exception e) {
            LOGGER.error("", e);
        }
    }

    public void init() throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
//        connectionFactory.setHost("192.168.99.100");
//        connectionFactory.setPort(32780);
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        //        创建连接
        Connection connection = connectionFactory.newConnection();
        //        在连接上建立渠道
        channel = connection.createChannel();
    }

    public void registerQueue(String... queueNames) {
        try {
            for (String queueName : queueNames) {
                channel.queueDeclare(queueName, false, true, true, (Map) null);
            }
            channel.addConfirmListener(new ConfirmListener() {
                @Override
                public void handleAck(long deliveryTag, boolean multiple) throws IOException {

                }

                @Override
                public void handleNack(long deliveryTag, boolean multiple) throws IOException {

                }
            });
            channel.addReturnListener(new ReturnListener() {
                @Override
                public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {

                }
            });
            //延迟队列
            //到期马上丢弃
//            Map<String, Object> args = new HashMap<String, Object>();
//            args.put("x-message-ttl", 6000);
//            channel.queueDeclare("", false, true, true, (Map) args);

            //投递之前进行过期判断
//            AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
//            builder.expiration("6000");
//            AMQP.BasicProperties properties = builder.build();
//            channel.basicPublish(exchangeName, routingKey, mandatory, properties, "msg body".getBytes());
        } catch (Exception e) {
            LOGGER.error("注册mq队列失败", e);
        }
    }

    public void registerExchange(String... exchangeNames) {
        try {
            for (String exchangeName : exchangeNames) {
                //        channel.exchangeBind("destExchanger","sourceExchanger","routingKey");
                //        声明渠道名和模式
                channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
            }
        } catch (Exception e) {
            LOGGER.error("", e);
        }
    }

    public void bind(String queue,String exchange,String routingKey) {
        try {
            channel.queueBind(queue,exchange,routingKey);
        } catch (Exception e) {
            LOGGER.error("", e);
        }
    }

    public void bindExchange(String dest,String source,String routingKey) {
        try {
            channel.exchangeBind(dest,source,routingKey);
        } catch (Exception e) {
            LOGGER.error("", e);
        }
    }

    public void consume(String queue,Consumer consumer) {
        try {
            channel.basicConsume(queue,consumer);
        } catch (Exception e) {
            LOGGER.error("", e);
        }
    }

    public Channel getChannel() {
        return channel;
    }

    public static void main(String[] args) throws IOException {
        MqClient mqClient = new MqClient();
        mqClient.registerQueue("q1");

        mqClient.registerQueue("q2");
        mqClient.registerExchange("x2");
        mqClient.bind("q2","x2","r2");

        mqClient.registerQueue("q3");
        mqClient.registerExchange("x3");
        mqClient.registerExchange("x4");
        mqClient.bind("q3","x3","r3");
        mqClient.bind("q2","x3","r2");
        mqClient.bindExchange("x4","x3","r4");


        Channel channel = mqClient.getChannel();

        channel.basicConsume("q1",new MqHandler(channel));
        channel.basicConsume("q2",new MqHandler(channel));
        channel.basicConsume("q3",new MqHandler(channel));
    }
}